package com.ehualu.liaocheng

/**
  * @Author: 吴敬超
  * @Date: 2019/9/7 21:18
  *
  *
  *        全市，没有电警过滤，按照卡口坐标计算的距离，然后计算速度
  */


import java.text.SimpleDateFormat

import com.ehualu.liaocheng.until.{DBConnection, OracleUtills}
import com.typesafe.config.ConfigFactory
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
import org.apache.spark.streaming.kafka010.KafkaUtils
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.{Minutes, StreamingContext}

import scala.collection.mutable.ArrayBuffer


object realTimekafka11 {


  def getDifferentSecondsu(start: String, end: String): Long = {

    val fm = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
    val startLong = fm.parse(start).getTime
    val endLong = fm.parse(end).getTime

    val seconds = (endLong - startLong) / 1000
    seconds
  }

  val getDifferentSeconds = getDifferentSecondsu _

  def main(args: Array[String]): Unit = {

    //    var conn = DBConnection.getehldb();

    //KAFKA安全认证
//    security.security.run()

    //KAFKA安全认证
    //    security.run()
    //创建SparkConf，如果将任务提交到集群中
    val conf = new SparkConf().setAppName("realTimeMonitoringtest2")
    //          .setMaster("local[2]")
    //创建一个StreamingContext，批次时长为2秒
    //    val ssc = new StreamingContext(conf, Seconds(30))
    val ssc = new StreamingContext(conf, Minutes(1))

    ssc.sparkContext.setLogLevel("WARN")
    //配置工厂 获取kakfa配置信息
    val load = ConfigFactory.load("kafka")

    //消费卡夫卡参数配置
    val kafkaParams = Map[String, Object](
      ///brokers 地址
      "bootstrap.servers" -> load.getString("kafka.brokers"),
      //指定该 consumer 将加入的消费组
      "group.id" -> "odoafng12344",
      //指定序列化类
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      // 是否开启自动提交 offset
      "enable.auto.commit" -> (false: java.lang.Boolean),
      "heartbeat.interval.ms" -> new Integer(100),
      "session.timeout.ms" -> new Integer(20000)
    )

    //主题列表
    val topics = Array(load.getString("kafka.topics"))

    //直连方式消费
    val stream: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](ssc,
      PreferConsistent,
      //订阅的策略（可以指定用正则的方式读取topic，比如my-ordsers-.*）
      Subscribe[String, String](topics, kafkaParams))


    //3715000530-3715000545
    val WindowsData: DStream[(String, (String, String, String, String))] = stream.map(

      line => {

        //        println("777777777777value777777777777：" + line.value())


        val arr: Array[String] = line.value().split(",")
        //      时间
        val sj = arr(3).substring(0, 19)

        //      号牌
        val hphm = arr(5)

        //      卡口编号
        val kkbh = arr(1)

        //        卡口名称
        val kkmc = arr(2)

        var dataid = arr(0)


        var kkbhi = kkbh.toLong
        (hphm, (kkbh, sj, kkmc, dataid))
      }


    ).window(Minutes(10))
      .filter(!_._1.toString.equals("车牌"))
    //      .filter(strcomp(_._2._3))
    //      .filter(
    //
    //      e => {
    //
    //        //        (!e._1.toString.equals("车牌")) || e._2._3.contains("东昌路") || e._2._3.contains("柳园路") || (!e._2._3.contains("电警"))
    //
    //        (!e._1.toString.equals("车牌")) && (!e._2._3.contains("电警"))
    //      }
    //
    //
    //    )
    //      .filter(_._2._1.toLong > "3715000530".toLong)
    //      .filter(_._2._1.toLong < "3715000545".toLong)
    //      .filter(!_._2._3.contains("电警"))
    //      .window(Seconds(60))
    //      .window(Minutes(10), Seconds(10))
    //      .window(Minutes(5))

    //    val pvWindow = data.reduceByKeyAndWindow(_ + _, _ - _, Minutes(60), Minutes(10))

    //过滤窗口数据
    val filterHphmGroupedRDD = WindowsData.groupByKey()
      .filter(e => {

        e._2.size > 1

      })

    //    遍历批次
    //    filterHphmGroupedRDD.foreachRDD(rdd => {
    //
    //
    //          rdd.foreachPartition(t => {
    //
    //            t.foreach(println)
    //
    //
    //          })
    //
    //
    //        })

    //
    //    -------------------------------------------
    //    Time: 1567561488000 ms
    //      -------------------------------------------
    //    (鲁P1P118,ArrayBuffer((3715000121,2019-09-04 09:44:32), (3715000122,2019-09-04 09:44:26), (3715000736,2019-09-04 09:43:54)))
    //    (鲁P9166W,ArrayBuffer((37150002336,2019-09-04 09:44:20), (37150002335,2019-09-04 09:44:25)))
    //    (鲁P77R03,ArrayBuffer((3715000145,2019-09-04 09:44:25), (3715000147,2019-09-04 09:44:09)))
    //    (鲁PG1665,ArrayBuffer((3715000650,2019-09-04 09:44:44), (3715000651,2019-09-04 09:44:08)))
    //    (鲁P27P77,ArrayBuffer((3715000059,2019-09-04 09:44:20), (3715000060,2019-09-04 09:44:17)))
    //    (鲁AL3C79,ArrayBuffer((3715000176,2019-09-04 09:43:54), (3715001006,2019-09-04 09:43:20)))
    //    (鲁P302A5,ArrayBuffer((3715000754,2019-09-04 09:44:41), (3715000755,2019-09-04 09:44:38)))
    //    (鲁PSZ879,ArrayBuffer((3715000765,2019-09-04 09:44:08), (3715000764,2019-09-04 09:44:14)))
    //    (鲁P17M60,ArrayBuffer((3715000160,2019-09-04 09:43:57), (3715000159,2019-09-04 09:44:01)))
    //    (鲁P57913,ArrayBuffer((3715000290,2019-09-04 09:44:25), (3715000291,2019-09-04 09:44:20)))
    //    ...
    //
    //    -------------------------------------------
    //    Time: 1567561490000 ms
    //      -------------------------------------------
    //    (鲁P1P118,ArrayBuffer((3715000121,2019-09-04 09:44:32), (3715000122,2019-09-04 09:44:26), (3715000736,2019-09-04 09:43:54)))
    //    (鲁P9166W,ArrayBuffer((37150002336,2019-09-04 09:44:20), (37150002335,2019-09-04 09:44:25)))
    //    (鲁P77R03,ArrayBuffer((3715000145,2019-09-04 09:44:25), (3715000147,2019-09-04 09:44:09)))
    //    (鲁PG1665,ArrayBuffer((3715000650,2019-09-04 09:44:44), (3715000651,2019-09-04 09:44:08)))
    //    (鲁P27P77,ArrayBuffer((3715000059,2019-09-04 09:44:20), (3715000060,2019-09-04 09:44:17)))
    //    (鲁AL3C79,ArrayBuffer((3715000176,2019-09-04 09:43:54), (3715001006,2019-09-04 09:43:20)))
    //    (鲁P302A5,ArrayBuffer((3715000754,2019-09-04 09:44:41), (3715000755,2019-09-04 09:44:38)))
    //    (鲁PSZ879,ArrayBuffer((3715000765,2019-09-04 09:44:08), (3715000764,2019-09-04 09:44:14)))
    //    (鲁P17M60,ArrayBuffer((3715000160,2019-09-04 09:43:57), (3715000159,2019-09-04 09:44:01)))
    //    (鲁P57913,ArrayBuffer((3715000290,2019-09-04 09:44:25), (3715000291,2019-09-04 09:44:20)))
    //    ...

    val roadTimeRDD = filterHphmGroupedRDD.map(iter => {
      //            println("7777777777777777777777777777777777777777777")
      //      println("55555555车牌55555555:" + iter._1)


      var array = ArrayBuffer[(String, String, String, String)]()
      val iterator = iter._2.iterator
      //取出每个key 的对应的信息
      while (iterator.hasNext) {

        //        Iterator类的next( )方法在同一循环中不能出现两次。
        //        println(iterator.next())
        array.append(iterator.next())

      }

      val sortedTupleArray: ArrayBuffer[(String, String, String, String)] = array.sortBy(_._2)


      //      println(sortedTupleArray)
      val size: Int = sortedTupleArray.size

      //      println("444444size44444:" + size)
      var rearray = ArrayBuffer[(String, Long)]()
      for (i <- (0 to size - 2)) {

        var kkbh1 = sortedTupleArray(i)._1
        var kkbh2 = sortedTupleArray(i + 1)._1
        var kksj1 = sortedTupleArray(i)._2
        var kksj2 = sortedTupleArray(i + 1)._2


        var databt1 = sortedTupleArray(i)._4
        var databt2 = sortedTupleArray(i + 1)._4
        //        println("****kkmc******" + sortedTupleArray(i)._3)
        //        println("****kkbh1******" + kkbh1)
        //        println("****kkbh2******" + kkbh2)
        //        println("****kksj1******" + kksj1)
        //        println("****kksj2******" + kksj2)

        //        println("*****路段名称**:" + sortedTupleArray(i)._3)
        //
        //
        //        println("*****路段编号**:" + kkbh1 + "_" + kkbh2)
        //        println("*****数据编号**：" + databt1 + "_" + databt2)
        //这辆车前后通过的卡口不是一个。前后是同一个卡口就过滤
        if (kkbh1 != kkbh2) {
          var ldbh = kkbh1 + "_" + kkbh2
          var ldsj = getDifferentSeconds(kksj1, kksj2)

          //                    两个卡口在同一个位置，通过时间会很短，过滤
          if (ldsj > 2) {


            var ldyz = (ldbh, ldsj)
            rearray.append(ldyz)
          }


        }

      }
      rearray

    }

    )

    //    roadTimeRDD.print()


    val notNullRoadRDD = roadTimeRDD.filter(e => {
      e.size > 0
    })

    println("8888888888888888888888888888888888888")


    val luDuanRdd = notNullRoadRDD.flatMap(x => x).groupByKey()

    val luduantjrdd = luDuanRdd.map(

      line => {
        println("&&&&&&&&路段编号&&&&&&&&&" + line._1)

        println("&&&&&&&&line._2&&&&&&&&&" + line._2.toString())

        var kkbh1 = line._1.split("_")(0)
        var kkbh2 = line._1.split("_")(1)

        var ydzs = ""

        //        卡口经纬度直线距离(m)
        var kkjl = OracleUtills.dissize(kkbh1, kkbh2) * 1000


        //        卡口间的距离大于0才计算
        if (kkjl > 0) {
          //        line._1
          var sjnum = line._2.iterator.size

          println("*******sjnum******" + sjnum)

          val sjiterator = line._2.iterator
          //        var sumsj: Long = 0

          var sumHs: Double = 0
          //取出每个key 的对应的信息
          while (sjiterator.hasNext) {

            var Hsj = sjiterator.next()

            var everyshisu = kkjl / Hsj


            //          速度 m/s   km/h   由于精度问题容易算出0 或无穷大来
            sumHs = sumHs + everyshisu


          }
          var ldpjss = sumHs / sjnum

          println("********路段平均时速 m/s***************：" + ldpjss)


          var ldpjsh = ldpjss * 3.6
          if (ldpjsh < 10) {

            ydzs = "拥堵"
            OracleUtills.writeora2(line._1, ldpjsh.toString, ydzs)
          } else if ((20 > ldpjsh) && (ldpjsh >= 10)) {

            ydzs = "缓行"
            OracleUtills.writeora2(line._1, ldpjsh.toString, ydzs)
          } else if ((ldpjsh < 80) && (ldpjsh >= 20)) {

            ydzs = "正常"
            OracleUtills.writeora2(line._1, ldpjsh.toString, ydzs)

          }


          (line._1, ldpjss)

        }

      }
    )


    //    println("99999999999999999999999999999999999999999")
    luduantjrdd.print()


    ssc.start()
    //等待退出
    ssc.awaitTermination()


    //    DBConnection.getcon(2)
    var conn = DBConnection.getcon(2)
    var conn1 = DBConnection.getcon(1)
    if (conn != null) {

      conn.close()

    }

    if (conn1 != null) {

      conn1.close()

    }


  }

}
